package com.baishancloud.log.live.spider

import com.baishancloud.log.common.util.{StringUtil, TimeUtil}
import com.baishancloud.log.format.java.util.IPLocator
import com.baishancloud.log.format.scala.SpiderOrigin
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}


class SpiderLogRecordParse(parameterTool: ParameterTool) extends ProcessFunction[SpiderOrigin, SpiderLog] {

  val LOGGER: Logger = LoggerFactory.getLogger(classOf[SpiderLogRecordParse])

  override def processElement(value: SpiderOrigin, ctx: ProcessFunction[SpiderOrigin, SpiderLog]#Context, out: Collector[SpiderLog]): Unit = {
    val ipInfo = IPLocator.lookup(value.clientIp())

    var isp = ipInfo.getIsp.split("/")(0)
    if (isp.equals("other")) isp = "其他"
    if (isp.contains("广电")) {
      isp = "广电"
    } else if (isp.contains("华数")) {
      isp = "华数"
    }
    val country = if (ipInfo.getCountry.equals("other")) "其他" else ipInfo.getCountry
    var province = if (ipInfo.getProvince.equals("other")) "其他" else ipInfo.getProvince
    if ((country.equals("中国") || country.startsWith("CHINA")) && province.equals("其他")) province = "中国其他地区"

    val logRecord: SpiderLog = SpiderLog(
      TimeUtil.timeAtString(value.datetimeOfUTC().toEpochSecond, parameterTool.getLong(timeAtSize, 1) * 60),
      StringUtil.subString(value.vhost(), 32),
      StringUtil.subString(country, 32),
      StringUtil.subString(province, 32),
      StringUtil.subString(isp, 32),
      1,
      value.uuid()
    )
    out.collect(logRecord)
  }
}
